package com.tjf.myjetpackdemo.RxObserverModel;

import android.os.Handler;
import android.os.Looper;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author: tjf
 * @date: 2023-06-08
 * @desc：  观察者的线程
 */
public class ObservableObserveOn<T> extends AbstractObservableWithUpstream<T, T> {


    protected ObservableObserveOn(ObservableSource<T> source) {
        super(source);

    }

    @Override
    protected void subscribeActual(final Observer<T> observer) {
        final SubscribeOnObserver<T> onObserver = new SubscribeOnObserver<>(observer);
        source.subscribe(onObserver);
    }

    static final class SubscribeOnObserver<T> implements Observer<T> {

        final Observer<T> actual;
        private Handler  handler;

        SubscribeOnObserver(Observer<T> actual) {
            this.actual = actual;
            handler=new Handler(Looper.getMainLooper());
        }

        @Override
        public void onSubscribe() {
            actual.onSubscribe();
        }

        @Override
        public void onNext(final T t) {
            handler.post(new Runnable() {
                @Override
                public void run() {
                    actual.onNext(t);
                }
            });
        }

        @Override
        public void onError(Throwable throwable) {
            actual.onError(throwable);
        }

        @Override
        public void onComplete() {
            actual.onComplete();
        }
    }
}
